package util

import (
	"fmt"
	"reflect"
	"sync"
)

var EventBus BusIf

// InitEventBus 初始化队列消息
func InitEventBus() {
	EventBus = NewEventBus(10000)
}

// BusIf 接口包含了全局的消息总线（bus）行为，包括订阅（subscribe）、发布（publish）和控制（control）功能。
type BusIf interface {
	// BusController 业务控制器
	BusController
	// BusSubscriber 业务订阅功能
	BusSubscriber
	// BusPublisher 业务消息发布
	BusPublisher
}

// BusController 定义总线控制行为（检查处理程序的存在、同步）
type BusController interface {
	HasCallback(topic string) bool
}

// BusSubscriber 定义与订阅相关的总线行为
type BusSubscriber interface {
	// DefaultSubscribe  通用队列,全局共用
	DefaultSubscribe(topic string, fn interface{}) error
	// DefaultSubscribeAsync 通用异步队列,全局调用
	DefaultSubscribeAsync(topic string, fn interface{}) error
	// Subscribe 独立队列,每个topic一个队列
	Subscribe(topic string, fn interface{}) error
	// SubscribeAsync 独立队列,异步实现。每个topic一个队列
	SubscribeAsync(topic string, fn interface{}) error
	// Unsubscribe 退订事件
	Unsubscribe(topic string, handler interface{}) error
}

// BusPublisher 定义与发布相关的总线行为
type BusPublisher interface {
	// DefaultPublish 默认队列,全局通用
	DefaultPublish(topic string, args ...interface{})
	// DefaultPublishAsync 默认队列,异步执行，全局通用
	DefaultPublishAsync(topic string, args ...interface{})
	// Publish 独立队列,每个topic作为一个单独的队列
	Publish(topic string, args ...interface{})
	// PublishAsync 独立队列,异步执行,每个topic作为一个队列
	PublishAsync(topic string, args ...interface{})
}

// EventBusImp 用于处理程序和回调的框。
type EventBusImp struct {
	// 默认处理器
	defaultHandlers map[string][]*EventHandler
	// 默认chanel通道队列
	defaultChannel chan []interface{}
	// 默认锁机制(互斥锁)
	defaultLock sync.Mutex

	topicHandlers map[string]*topicHandler
	// 通道缓冲大小
	queueSize int
}

// EventHandler 事件处理器
type EventHandler struct {
	// 回调函数
	callBack reflect.Value
	// 异步是否
	async bool
}

// topicHandler 目标处理器
type topicHandler struct {
	name     string             // 通道名称
	handlers []*EventHandler    // 通道订阅列表
	chnl     chan []interface{} // 通道队列
	lock     sync.Mutex         // 对处理器进行上锁
}

// loop 循环永不退出等待channel通道接受消息
func (th *topicHandler) loop(bus *EventBusImp) {
	// 开启协程
	go func() {
		// 无线循环
		for {
			select {
			case args := <-th.chnl:
				topic := args[0].(string)
				bus.Publish(topic, args[1:]...)
			}
		}
	}()
}

// removeHandler 删除处理器
func (th *topicHandler) removeHandler(topic string, callback reflect.Value) int {
	// 开启互斥锁
	th.lock.Lock()
	defer th.lock.Unlock()

	idx := -1
	for i, handler := range th.handlers {
		// 判断返回函数的动态类型或者返回的指针对应的内存地址
		if handler.callBack.Type() == callback.Type() &&
			handler.callBack.Pointer() == callback.Pointer() {
			idx = i
			break
		}
	}
	// 判断 处理器列表的长度
	nSize := len(th.handlers)
	if 0 > idx || idx >= nSize {
		return -1
	}

	th.handlers = append(th.handlers[:idx], th.handlers[idx+1:]...)
	return 0
}

// NewEventBus 返回带有空处理程序的新EventBusImp。
func NewEventBus(queSize int) BusIf {
	// 对EventBusImp进行实例化
	eb := &EventBusImp{
		defaultHandlers: make(map[string][]*EventHandler),
		defaultChannel:  make(chan []interface{}, queSize),
		defaultLock:     sync.Mutex{},
		topicHandlers:   make(map[string]*topicHandler),
		queueSize:       queSize,
	}
	// 开启循环协程永不退出接收对接消息
	eb.loop()
	return BusIf(eb)
}

// loop 协程永不退出获取channel通道消息
func (bus *EventBusImp) loop() {
	// 建立一个协程
	go func() {
		// 协程永不退出
		for {
			select {
			case args := <-bus.defaultChannel:
				topic := args[0].(string)
				bus.Publish(topic, args[1:]...)
			}
		}
	}()
}

// -------------------------------------- 订 阅 事 件 -------------------------------------------------------------------
// Subscribe 订阅一个主题 Return:如果fn不是函数返回错误
// 全集公用队列

func (bus *EventBusImp) DefaultSubscribe(topic string, fn interface{}) error {
	// 如果fn类型不是函数类型则报错
	if !(reflect.TypeOf(fn).Kind() == reflect.Func) {
		return fmt.Errorf("%s is not of type reflect.Func", reflect.TypeOf(fn).Kind())
	}
	// 开启互斥锁
	bus.defaultLock.Lock()
	defer bus.defaultLock.Unlock()

	handler := &EventHandler{
		callBack: reflect.ValueOf(fn),
		async:    false,
	}

	// 新增处理操作，添加到默认公共的处理器
	bus.defaultHandlers[topic] = append(bus.defaultHandlers[topic], handler)
	return nil
}

// SubscribeAsync 使用异步回调订阅主题.
// 事务性决定了主题的后续回调是否
// 串行运行（true）或并发运行（false）,如果`fn`不是函数，则返回错误。

func (bus *EventBusImp) DefaultSubscribeAsync(topic string, fn interface{}) error {
	// 判断调用类型
	if !(reflect.TypeOf(fn).Kind() == reflect.Func) {
		return fmt.Errorf("%s is not of type reflect.Func", reflect.TypeOf(fn).Kind())
	}
	// 开启互斥锁
	bus.defaultLock.Lock()
	defer bus.defaultLock.Unlock()
	handler := &EventHandler{
		callBack: reflect.ValueOf(fn),
		async:    true,
	}
	bus.defaultHandlers[topic] = append(bus.defaultHandlers[topic], handler)
	return nil
}

// addHandler 为每个独立topic添加用户处理函数
func (bus *EventBusImp) addHandler(topic string, handler *EventHandler) error {
	var tHandler *topicHandler
	var ok bool
	if tHandler, ok = bus.topicHandlers[topic]; !ok {
		// map 不存在key则新建
		tHandler = &topicHandler{
			name: topic,
			chnl: make(chan []interface{}, bus.queueSize),
			lock: sync.Mutex{},
		}
		bus.topicHandlers[topic] = tHandler
		tHandler.loop(bus)
	}
	// 加互斥锁
	tHandler.lock.Lock()
	defer tHandler.lock.Unlock()
	tHandler.handlers = append(tHandler.handlers, handler)
	return nil
}

// Subscribe 订阅一个主题
func (bus *EventBusImp) Subscribe(topic string, fn interface{}) error {
	// 如果fn类型不是函数类型则报错
	if !(reflect.TypeOf(fn).Kind() == reflect.Func) {
		return fmt.Errorf("%s is not of type reflect.Func", reflect.TypeOf(fn).Kind())
	}

	handler := &EventHandler{
		callBack: reflect.ValueOf(fn),
		async:    false,
	}
	return bus.addHandler(topic, handler)
}

// SubscribeAsync 异步订阅主图
func (bus *EventBusImp) SubscribeAsync(topic string, fn interface{}) error {
	// 如果fn类型不是函数类型则报错
	if !(reflect.TypeOf(fn).Kind() == reflect.Func) {
		return fmt.Errorf("%s is not of type reflect.Func", reflect.TypeOf(fn).Kind())
	}
	handler := &EventHandler{
		callBack: reflect.ValueOf(fn),
		async:    true,
	}
	return bus.addHandler(topic, handler)
}

// -------------------------------------------退 订 事 件-----------------------------------------------------------------

// Unsubscribe 删除为主题定义的回调。如果没有订阅主题的回调，则返回错误。
func (bus *EventBusImp) Unsubscribe(topic string, handler interface{}) error {
	var ret int
	if ret = bus.defaultRemoveHandler(topic, reflect.ValueOf(handler)); ret == 0 {
		return nil
	}
	if tHandler, ok := bus.topicHandlers[topic]; ok {
		if ret = tHandler.removeHandler(topic, reflect.ValueOf(handler)); ret == 0 {
			return nil
		}
	}
	return fmt.Errorf("topic %s not exist", topic)
}

// -------------------------------------------分 发 事 件-----------------------------------------------------------------

// DefaultPublish 分发全局队列
func (bus *EventBusImp) DefaultPublish(topic string, args ...interface{}) {
	// 开启互斥锁
	bus.defaultLock.Lock()
	defer bus.defaultLock.Unlock()

	if handlers, ok := bus.defaultHandlers[topic]; ok {
		bus.doPublish(handlers, args...)
	}
}

// DefaultPublishAsync 异步分发全局队列
func (bus *EventBusImp) DefaultPublishAsync(topic string, args ...interface{}) {
	// 定义一个泛型的数组
	var passedArguments []interface{}
	// 初始化一个泛型的数组长度为args长度+1,因为要给第一个赋值
	passedArguments = make([]interface{}, len(args)+1)
	// 第一个参数，赋值标题
	passedArguments[0] = topic
	// 循环参数,从1开始把参数放入的数组
	for i, v := range args {
		passedArguments[i+1] = v
	}
	// 返回给channel通道
	bus.defaultChannel <- passedArguments
}

// Publish 单独处理器
func (bus *EventBusImp) Publish(topic string, args ...interface{}) {
	if topicHandler, ok := bus.topicHandlers[topic]; ok {
		bus.doPublish(topicHandler.handlers, args...)
	}
}

// PublishAsync 异步分发事件
func (bus *EventBusImp) PublishAsync(topic string, args ...interface{}) {
	if topicHandler, ok := bus.topicHandlers[topic]; ok {
		var passedArguments []interface{}
		passedArguments = make([]interface{}, len(args)+1)
		passedArguments[0] = topic
		for i, v := range args {
			passedArguments[i+1] = v
		}
		topicHandler.chnl <- passedArguments
	} else {
		fmt.Errorf("topic %s doesn't exist", topic)
	}

}

// doPublish 执行为主题定义的回调。任何额外的参数都将被转移到回调函数中。
func (bus *EventBusImp) doPublish(handlers []*EventHandler, args ...interface{}) {
	// 判断处理器列表长度大于
	if len(handlers) > 0 {
		// 在迭代过程中，可以通过removeHandler和Unsubscribe来更改处理程序切片，
		// 因此，制作一个副本并迭代复制的切片。
		// 初始化一个长度和容量都为处理个数的切片
		copyHandlers := make([]*EventHandler, len(handlers))
		copy(copyHandlers, handlers)
		for _, handler := range copyHandlers {
			if handler != nil {
				if handler.async {
					go bus.doCallFunc(handler, args...)
				} else {
					bus.doCallFunc(handler, args...)
				}
			}
		}
	}
}

// doCallFunc 回调函数
func (bus *EventBusImp) doCallFunc(handler *EventHandler, args ...interface{}) {
	funcType := handler.callBack.Type()
	passArguments := make([]reflect.Value, len(args))
	for i, v := range args {
		if v == nil {
			passArguments[i] = reflect.New(funcType.In(i))
		} else {
			passArguments[i] = reflect.ValueOf(v)
		}
	}
	handler.callBack.Call(passArguments)
}

// defaultRemoveHandler 实现接口的方法
func (bus *EventBusImp) defaultRemoveHandler(topic string, callback reflect.Value) int {
	// 开启互斥锁
	bus.defaultLock.Lock()
	defer bus.defaultLock.Unlock()

	// 判断订阅主题是否存在
	handlers, ok := bus.defaultHandlers[topic]
	if !ok {
		return -1
	}
	idx := -1

	// 判断 EventHandler里面callback 类型或者返回的指针知否相同
	for i, handler := range handlers {
		if handler.callBack.Type() == callback.Type() &&
			handler.callBack.Pointer() == callback.Pointer() {
			idx = i
			break
		}
	}
	// 查看handler的长度
	nSize := len(handlers)
	if 0 > idx || idx >= nSize {
		return -1
	}
	copy(handlers[idx:], handlers[idx+1:])
	handlers[nSize-1] = nil
	return 0
}

// HasCallback 如果存在订阅该主题的任何回调，则返回true。
func (bus *EventBusImp) HasCallback(topic string) bool {
	// 开启 互斥锁
	bus.defaultLock.Lock()
	defer bus.defaultLock.Unlock()

	handlers, ok := bus.defaultHandlers[topic]
	if ok {
		return len(handlers) > 0
	} else {
		tHandler, ok := bus.topicHandlers[topic]
		if ok {
			return len(tHandler.handlers) > 0
		}
	}
	return false
}
